/*
 * Project Name spring-boot-tools
 * File Name AmppGateway
 * Package Name com.huxiaosu.tools.cmpp
 * Create Time 2024/4/18
 * Create by name：liujie
 */
package com.huxiaosu.tools.cmpp;

import com.huxiaosu.tools.cmpp.codec.CodecFactory;
import com.huxiaosu.tools.cmpp.config.CmppGatewayClient;
import com.huxiaosu.tools.cmpp.config.CmppGatewayConfig;
import com.huxiaosu.tools.cmpp.filter.ConnectionFilter;
import com.huxiaosu.tools.cmpp.filter.keepalive.KeepAliveMessageFactoryImpl;
import com.huxiaosu.tools.cmpp.cache.CacheService;
import com.huxiaosu.tools.cmpp.action.MessageHandler;
import com.huxiaosu.tools.common.executors.ExecutorsUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.mina.core.service.IoAcceptor;
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.executor.ExecutorFilter;
import org.apache.mina.filter.keepalive.KeepAliveFilter;
import org.apache.mina.filter.logging.LoggingFilter;
import org.apache.mina.transport.socket.nio.NioSocketAcceptor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.web.context.WebServerInitializedEvent;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.event.EventListener;
import org.springframework.core.annotation.Order;
import org.springframework.scheduling.annotation.Async;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.concurrent.Executors;

/**
 * Description
 *
 * @author liujie
 * @date 2024/4/18 23:00
 */
@Slf4j
@Configuration(proxyBeanMethods = false)
public class CmppGateway {

    @Autowired
    private CmppGatewayConfig cmppGatewayConfig;

    @Autowired
    private CacheService cacheService;

    @Async
    @Order
    @EventListener(WebServerInitializedEvent.class)
    public void afterStart(WebServerInitializedEvent event) {
        try {
            IoAcceptor acceptor = new NioSocketAcceptor();
            acceptor.getFilterChain().addLast("logger", new LoggingFilter());
            acceptor.getFilterChain().addLast("myfliter", new ConnectionFilter());
            // 指定编码过滤器
            acceptor.getFilterChain().addLast("codec", new ProtocolCodecFilter(new CodecFactory("UTF-8")));
            acceptor.getFilterChain().addLast("threadPool", new ExecutorFilter(ExecutorsUtil.createExecutors()));
            // 发送缓冲区10M
            acceptor.getSessionConfig().setReadBufferSize(2048 * 5000);
            // acceptor.getSessionConfig().set
            KeepAliveMessageFactoryImpl kamfi = new KeepAliveMessageFactoryImpl();
            KeepAliveFilter kaf = new KeepAliveFilter(kamfi, IdleStatus.BOTH_IDLE);
            /** 是否回发 */
            kaf.setForwardEvent(true);
            acceptor.getFilterChain().addLast("heart", kaf);
            // 指定业务逻辑处理器
            acceptor.setHandler(new MessageHandler());
            // 设置端口号
            acceptor.setDefaultLocalAddress(new InetSocketAddress(cmppGatewayConfig.getPort()));

            acceptor.getSessionConfig().setIdleTime(IdleStatus.BOTH_IDLE, 600);

            acceptor.bind();// 启动监听

            log.info("< CMPP server start success, port is ={}>", cmppGatewayConfig.getPort());

            loadClient();
        }
        catch (Throwable e) {
            log.warn(" CMPP server start error", e);
        }
    }

    private void loadClient() {
        // 这里可以在数据库加载
        List<CmppGatewayClient> clients = cmppGatewayConfig.getClients();
        clients.forEach(item -> {
            cacheService.put(item.getSpIp(), item);
            log.info("< load client success, ip is ={} serviceId is = {} >", item.getSpIp(), item.getServiceId());
        });
    }

}
